package reactor;

import org.junit.Test;
import reactor.core.publisher.Flux;
import util.Util;

import java.util.Arrays;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;


/**
 * @author yding5
 */
public class FluxTest {

    @Test
    public void testJust() {
        Flux.just(1, 2, 3).subscribe(Util::print);
    }

    @Test
    public void testFromIterable() {
        Flux.fromIterable(Arrays.asList(1, 2, 3)).subscribe(Util::print);
    }

    @Test
    public void testFromStream() {
        Flux.fromStream(Stream.of(1, 2, 3)).subscribe(Util::print);
    }

    /**
     * 编程式创建序列
     * 状态感知型下发元素
     */
    @Test
    public void testGenerate() {
        Flux.generate(() -> 0, (state, sink) -> {
            if (state > 10) {
                sink.complete();
            } else {
                Util.print("生产者生产内容：" + state);
                sink.next(state);
            }
            return ++state;
        }).subscribe(Util::print);

        /*for (int i = 0; i < 3; i++) {
            生产元素
        }*/
    }

    /**
     * 编程式创建序列
     * 无状态下发元素
     */
    @Test
    public void testCreateSync() {
        Flux.create(sink -> {
            for (int i = 0; i < 50; i++) {
                Util.print("发射了元素" + i);
                sink.next(i);
            }
            sink.complete();
        }).subscribe(Util::print);
    }

    /**
     * 编程式创建序列
     * 可以结合异步下发元素
     */
    @Test
    public void testCreateAsync() {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(50);
        new Thread(() -> {
            for (int i = 0; i < 50; i++) {
                queue.offer(i);
            }
        }).start();
        Flux.create(sink -> {
            try {
                Integer i;
                while ((i = queue.poll(100, TimeUnit.MILLISECONDS)) != null) {
                    sink.next(i);
                }
                sink.complete();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }).subscribe(Util::print);
    }
}
